internal/worker/worker.go (1)

123-135: Fix: failed retry results are being treated as successes (can panic downstream).

In processChunkWithRetry, an errored retry is stored in successMap and later appended to finalResults, so Poller.convertPollResultsToBlockData can dereference nil Data. Keep errors in a separate failed map and return them via failedResults.

Apply this diff:

-	// Separate successful and failed
-	successMap := make(map[string]rpc.GetFullBlockResult)
-	var failedBlocks []*big.Int
+	// Separate successful and failed
+	successMap := make(map[string]rpc.GetFullBlockResult)
+	failedMap := make(map[string]rpc.GetFullBlockResult)
+	var failedBlocks []*big.Int

 	for i, result := range results {
 		if i < len(chunk) {
 			if result.Error == nil {
 				successMap[chunk[i].String()] = result
 			} else {
-				failedBlocks = append(failedBlocks, chunk[i])
+				failedBlocks = append(failedBlocks, chunk[i])
+				failedMap[chunk[i].String()] = rpc.GetFullBlockResult{
+					BlockNumber: chunk[i],
+					Error:       result.Error,
+				}
 			}
 		}
 	}

 	// If only one block failed, retry once more
 	if len(failedBlocks) == 1 {
 		retryResults := fetchFunc(ctx, failedBlocks)
 		if len(retryResults) > 0 {
 			if retryResults[0].Error == nil {
 				successMap[failedBlocks[0].String()] = retryResults[0]
 			} else {
-				// Keep the error result
-				successMap[failedBlocks[0].String()] = rpc.GetFullBlockResult{
-					BlockNumber: failedBlocks[0],
-					Error:       retryResults[0].Error,
-				}
+				// Track as failed (do not add to successMap)
+				failedMap[failedBlocks[0].String()] = rpc.GetFullBlockResult{
+					BlockNumber: failedBlocks[0],
+					Error:       retryResults[0].Error,
+				}
 			}
+		} else {
+			failedMap[failedBlocks[0].String()] = rpc.GetFullBlockResult{
+				BlockNumber: failedBlocks[0],
+				Error:       fmt.Errorf("retry did not return result"),
+			}
 		}
 	} else if len(failedBlocks) > 1 {
@@
-		go func() {
+		go func() {
 			defer rwg.Done()
-			leftResults, _ := w.processChunkWithRetry(ctx, leftChunk, fetchFunc)
+			leftResults, leftFailed := w.processChunkWithRetry(ctx, leftChunk, fetchFunc)
 			// Add results to map
 			for _, r := range leftResults {
 				if r.BlockNumber != nil {
 					rwgMutex.Lock()
 					successMap[r.BlockNumber.String()] = r
 					rwgMutex.Unlock()
 				}
 			}
+			for _, fr := range leftFailed {
+				if fr.BlockNumber != nil {
+					rwgMutex.Lock()
+					failedMap[fr.BlockNumber.String()] = fr
+					rwgMutex.Unlock()
+				}
+			}
 		}()
@@
-		go func() {
+		go func() {
 			defer rwg.Done()
-			rightResults, _ := w.processChunkWithRetry(ctx, rightChunk, fetchFunc)
+			rightResults, rightFailed := w.processChunkWithRetry(ctx, rightChunk, fetchFunc)
 			// Add results to map
 			for _, r := range rightResults {
 				if r.BlockNumber != nil {
 					rwgMutex.Lock()
 					successMap[r.BlockNumber.String()] = r
 					rwgMutex.Unlock()
 				}
 			}
+			for _, fr := range rightFailed {
+				if fr.BlockNumber != nil {
+					rwgMutex.Lock()
+					failedMap[fr.BlockNumber.String()] = fr
+					rwgMutex.Unlock()
+				}
+			}
 		}()
@@
 	// Build final results in original order
 	var finalResults []rpc.GetFullBlockResult
 	var failedResults []rpc.GetFullBlockResult
 	for _, block := range chunk {
 		if result, ok := successMap[block.String()]; ok {
 			finalResults = append(finalResults, result)
+		} else if f, ok := failedMap[block.String()]; ok {
+			failedResults = append(failedResults, f)
 		} else {
 			// This should not happen as we have retried all failed blocks
 			failedResults = append(failedResults, rpc.GetFullBlockResult{
 				BlockNumber: block,
 				Error:       fmt.Errorf("failed to fetch block"),
 			})
 		}
 	}

Also applies to: 136-151, 162-190, 195-210

internal/orchestrator/committer.go (2)

191-217: Avoid tight loop when no blocks to commit.

Without interval pacing, this spins and floods logs when empty.

Apply this diff:

 			if len(blockDataToCommit) == 0 {
 				log.Debug().Msg("No block data to commit")
-				continue
+				time.Sleep(50 * time.Millisecond)
+				continue
 			}

---

219-236: Avoid tight publish loop.

Similarly, add a small idle backoff when there’s nothing to publish (implemented inside publish()).

Apply this diff in publish():

 	if len(blockData) == 0 {
-		return nil
+		time.Sleep(50 * time.Millisecond)
+		return nil
 	}

internal/orchestrator/poller.go (2)

123-131: Fix race: atomically check-and-mark processing to prevent double polling

Between isRangeProcessing and markRangeAsProcessing there’s a TOCTOU race; two goroutines can poll the same range concurrently. Do the check-and-set under one lock.

Apply:

-	// Check if already processing
-	if p.isRangeProcessing(rangeKey) {
-		return nil, ErrBlocksProcessing
-	}
-
-	p.markRangeAsProcessing(rangeKey)
-	defer p.unmarkRangeAsProcessing(rangeKey)
+	// Atomically check-and-mark as processing
+	if !p.tryMarkRangeAsProcessing(rangeKey) {
+		return nil, ErrBlocksProcessing
+	}
+	defer p.unmarkRangeAsProcessing(rangeKey)

And add:

+// tryMarkRangeAsProcessing atomically marks a range as processing.
+func (p *Poller) tryMarkRangeAsProcessing(rangeKey string) bool {
+	p.processingRangesMutex.Lock()
+	defer p.processingRangesMutex.Unlock()
+	if _, exists := p.processingRanges[rangeKey]; exists {
+		return false
+	}
+	// Presence of key denotes "processing"; waiters slice starts empty.
+	p.processingRanges[rangeKey] = []chan struct{}{}
+	return true
+}

---

131-134: Return the sentinel ErrNoNewBlocks so callers handle it correctly

poll() currently returns a generic error for “no valid block data,” but callers (Request/processBatch) special-case ErrNoNewBlocks. This causes unnecessary error logs.

-	if len(blockData) == 0 || highestBlockNumber == nil {
-		return nil, fmt.Errorf("no valid block data polled")
-	}
+	if len(blockData) == 0 || highestBlockNumber == nil {
+		return nil, ErrNoNewBlocks
+	}

internal/orchestrator/committer.go (2)

191-217: Avoid busy-looping when there’s nothing to commit

With interval pacing removed, this loop will spin and log constantly if no data is ready.

 			if len(blockDataToCommit) == 0 {
-				log.Debug().Msg("No block data to commit")
-				continue
+				log.Debug().Msg("No block data to commit")
+				time.Sleep(100 * time.Millisecond) // small backoff to avoid hot loop
+				continue
 			}

Optionally make the backoff configurable.

---

219-236: Publish loop can spin; backoff when no progress

publish() returns nil even when no blocks were published; add a progress check and brief sleep.

 func (c *Committer) runPublishLoop(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		default:
-			if c.commitToBlock.Sign() > 0 && c.lastPublishedBlock.Load() >= c.commitToBlock.Uint64() {
+			if c.commitToBlock.Sign() > 0 && c.lastPublishedBlock.Load() >= c.commitToBlock.Uint64() {
 				// Completing the publish loop if we've published more than commit to block
 				log.Info().Msgf("Committer reached configured toBlock %s, the last publish block is %d, stopping publishes", c.commitToBlock.String(), c.lastPublishedBlock.Load())
 				return
 			}
-			if err := c.publish(ctx); err != nil {
+			before := c.lastPublishedBlock.Load()
+			if err := c.publish(ctx); err != nil {
 				log.Error().Err(err).Msg("Error publishing blocks")
 			}
+			// Backoff if no progress
+			if c.lastPublishedBlock.Load() == before {
+				time.Sleep(100 * time.Millisecond)
+			}
 			go c.cleanupProcessedStagingBlocks(ctx)
 		}
 	}
 }

internal/orchestrator/work_mode_monitor.go (3)

49-63: Fix panic on send to closed channel + avoid holding lock while sending.

RegisterChannel can panic if ch is closed, and currently holds the write lock during send. Snapshot mode under lock, unlock, then do a safe non-blocking send with recover; also remove closed channels.

Apply this diff:

 func (m *WorkModeMonitor) RegisterChannel(ch chan WorkMode) {
-	m.channelsMutex.Lock()
-	defer m.channelsMutex.Unlock()
-
-	m.workModeChannels[ch] = struct{}{}
-	// Send current mode to the new channel only if it's not empty
-	if m.currentMode != "" {
-		select {
-		case ch <- m.currentMode:
-			log.Debug().Msg("Initial work mode sent to new channel")
-		default:
-			log.Warn().Msg("Failed to send initial work mode to new channel - channel full")
-		}
-	}
+	// Add channel under lock, snapshot current mode, then release
+	m.channelsMutex.Lock()
+	m.workModeChannels[ch] = struct{}{}
+	mode := m.currentMode
+	m.channelsMutex.Unlock()
+
+	// Send initial mode (non-blocking) outside lock; tolerate closed channels
+	if mode != "" {
+		func() {
+			defer func() {
+				if r := recover(); r != nil {
+					log.Warn().Msg("Failed to send initial work mode - channel closed; unregistering")
+					m.channelsMutex.Lock()
+					delete(m.workModeChannels, ch)
+					m.channelsMutex.Unlock()
+				}
+			}()
+			select {
+			case ch <- mode:
+				log.Debug().Msg("Initial work mode sent to new channel")
+			default:
+				log.Warn().Msg("Failed to send initial work mode to new channel - channel full")
+			}
+		}()
+	}
 }

---

81-118: Guard currentMode writes to avoid data races.

currentMode is read from RegisterChannel; protect writes here with the same mutex.

Apply this diff:

-		log.Info().Msgf("Work mode changing from %s to %s during startup", m.currentMode, newMode)
-		m.currentMode = newMode
+		log.Info().Msgf("Work mode changing from %s to %s during startup", m.currentMode, newMode)
+		m.channelsMutex.Lock()
+		m.currentMode = newMode
+		m.channelsMutex.Unlock()
 		m.updateWorkModeMetric(newMode)
 		m.broadcastWorkMode(newMode)
@@
-			if newMode != m.currentMode {
-				log.Info().Msgf("Work mode changing from %s to %s", m.currentMode, newMode)
-				m.currentMode = newMode
+			if newMode != m.currentMode {
+				log.Info().Msgf("Work mode changing from %s to %s", m.currentMode, newMode)
+				m.channelsMutex.Lock()
+				m.currentMode = newMode
+				m.channelsMutex.Unlock()
 				m.updateWorkModeMetric(newMode)
 				m.broadcastWorkMode(newMode)
 			}

---

120-135: Fix unsafe recover and illegal map mutation under RLock in broadcast.

recover in default branch never triggers; deleting from map while RLock held is unsafe. Snapshot channels, send with recover, then delete closed channels under write lock.

Apply this diff:

 func (m *WorkModeMonitor) broadcastWorkMode(mode WorkMode) {
-	m.channelsMutex.RLock()
-	defer m.channelsMutex.RUnlock()
-
-	for ch := range m.workModeChannels {
-		select {
-		case ch <- mode:
-			log.Debug().Msg("Work mode change notification sent")
-		default:
-			if r := recover(); r != nil {
-				log.Warn().Msg("Work mode notification dropped - channel closed")
-				delete(m.workModeChannels, ch)
-			}
-		}
-	}
+	// Snapshot channels to avoid holding write lock while sending
+	m.channelsMutex.RLock()
+	channels := make([]chan WorkMode, 0, len(m.workModeChannels))
+	for ch := range m.workModeChannels {
+		channels = append(channels, ch)
+	}
+	m.channelsMutex.RUnlock()
+
+	var toRemove []chan WorkMode
+	for _, ch := range channels {
+		closed := false
+		func() {
+			defer func() {
+				if r := recover(); r != nil {
+					closed = true
+				}
+			}()
+			select {
+			case ch <- mode:
+				log.Debug().Msg("Work mode change notification sent")
+			default:
+				log.Warn().Msg("Work mode notification dropped - channel full")
+			}
+		}()
+		if closed {
+			toRemove = append(toRemove, ch)
+		}
+	}
+
+	if len(toRemove) > 0 {
+		m.channelsMutex.Lock()
+		for _, ch := range toRemove {
+			delete(m.workModeChannels, ch)
+		}
+		m.channelsMutex.Unlock()
+		log.Warn().Int("removed", len(toRemove)).Msg("Removed closed work mode channels")
+	}
 }

configs/config.go (1)

343-351: Bug: Orchestrator chainBasedConfig writes to Main instead of Orchestrator.

This overwrites the wrong target; use Storage.Orchestrator.Clickhouse.

Apply this diff:

-	if Cfg.Storage.Main.Clickhouse != nil {
-		Cfg.Storage.Main.Clickhouse.ChainBasedConfig = orchestratorChainConfig
-	}
+	if Cfg.Storage.Orchestrator.Clickhouse != nil {
+		Cfg.Storage.Orchestrator.Clickhouse.ChainBasedConfig = orchestratorChainConfig
+	}